package com.raylu.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.raylu.realtime.bean.*;
import com.raylu.realtime.utils.KafkaSinkUtil;
import com.raylu.realtime.utils.KafkaSourceUtil;
import com.raylu.realtime.utils.PropertiesUtil;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Date;
import java.text.SimpleDateFormat;
import java.util.Properties;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2021/12/24.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
public class BaseLogApp2 {

    public static void main(String[] args) throws Exception {
        Properties prop = PropertiesUtil.load("config.properties");

        //TODO 1.基础环境设置
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点设置
        //TODO 2.1. 开启检查点
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //TODO 2.2. 设置检查点超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
        //TODO 2.3. job取消后，检查点是否保留
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //TODO 2.4. 设置状态后端
//        env.setStateBackend(new FsStateBackend(prop.getProperty("base.log.app.fsstatebackend.url")));
        //TODO 2.5. 设置hadoop使用的用户
//        System.setProperty("HADOOP_USER_NAME", "raylu");
        //TODO 3.设置重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(3)));
        //TODO 4.Kafka Source设置
        OutputTag<String> actionLogTag = new OutputTag<String>("ActionLogTag") {
        };
        OutputTag<String> startLogTag = new OutputTag<String>("StartLogTag") {
        };
        OutputTag<String> displayLogTag = new OutputTag<String>("DisplayLogTag") {
        };

        SingleOutputStreamOperator<String> resultOp = env
                .addSource(KafkaSourceUtil.getKafkaSource(prop.getProperty("base.log.app.kafka.topic"), prop.getProperty("base.log.app.kafka.group-id")))
                //TODO 5.数据格式转换 String -> JSONObject
                .map(r -> JSON.parseObject(r, LogBean.class))
                //TODO 6.新旧客户筛查
                .keyBy(r -> r.common.md)
                .process(new KeyedProcessFunction<String, LogBean, String>() {

                    private MapState<String, String> isNewDevMap;
                    private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        isNewDevMap = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("isNewDevMap", Types.STRING, Types.STRING));
                    }

                    @Override
                    public void processElement(LogBean value, KeyedProcessFunction<String, LogBean, String>.Context ctx, Collector<String> out) throws Exception {
                        if ("1".equals(value.common.is_new)) {
                            String mid = value.common.mid;
                            Long ts = value.ts;
                            String formatDate = sdf.format(new Date(ts));
                            if (isNewDevMap.get(mid) != null && isNewDevMap.get(mid).length() > 0) {
                                value.common.is_new = "0";
                            }else{
                                isNewDevMap.put(mid, formatDate);
                            }
                        }


                        //TODO 7. 分流
                        //TODO 7.1. 启动日志
                        if (value.start != null) {
                            StartBean startBean = value.start;
                            BeanUtils.copyProperties(startBean, value);//ts copy
//                            BeanUtils.copyProperties(startBean, value.common);//common copy
                            ctx.output(startLogTag, startBean.toString());
                        }
                        //TODO 7.2. 页面日志
                        else {
                            PageBean pageBean = value.page;
                          BeanUtils.copyProperties(pageBean, value);//ts copy
                          BeanUtils.copyProperties(pageBean, value.common);//common copy
                            out.collect(pageBean.toString());
                            //TODO 7.3. 曝光日志
                            if (value.displays != null && value.displays.size() > 0) {
                                for (DisplayBean displayBean : value.displays) {
                                    BeanUtils.copyProperties(displayBean, value.common);//common
                                    BeanUtils.copyProperties(displayBean, value.page);//page
                                    BeanUtils.copyProperties(displayBean, value);//ts
                                    ctx.output(displayLogTag, displayBean.toString());
                                }


                            }

                            //TODO 7.4. 活动日志
                            if (value.actions != null && value.actions.size() > 0) {
                                for (ActionBean actionBean : value.actions) {
                                    BeanUtils.copyProperties(actionBean, value.common);//common
                                    BeanUtils.copyProperties(actionBean, value.page);//page
                                    actionBean.log_ts = value.ts;
                                    ctx.output(actionLogTag, actionBean.toString());
                                }
                            }
                        }

                    }
                });
        //TODO 8. 输出
        //TODO 8.1. 页面日志输出
        resultOp.addSink(KafkaSinkUtil.getKafkaSink("DWD_PAGE_LOG"));
        //TODO 8.2. 启动日志输出
        DataStream<String> startOutput = resultOp.getSideOutput(startLogTag);
        startOutput.addSink(KafkaSinkUtil.getKafkaSink("DWD_START_LOG"));
        //TODO 8.3. 曝光日志输出
        DataStream<String> displayOutput = resultOp.getSideOutput(displayLogTag);
        displayOutput.addSink(KafkaSinkUtil.getKafkaSink("DWD_DISPLAY_LOG"));
        //TODO 8.4. 活动日志输出
        DataStream<String> actionOutput = resultOp.getSideOutput(actionLogTag);
        actionOutput.addSink(KafkaSinkUtil.getKafkaSink("DWD_ACTION_LOG"));

        resultOp.print("Page");
        startOutput.print("start");
        displayOutput.print("display");
        actionOutput.print("action");

        env.execute();
    }


}
